package com.hxq.demo1;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @author hxq
 * @date 2022/4/14 17:39
 */
//@Slf4j
public class CustomProducer {

    public static void main(String[] args) throws InterruptedException, ExecutionException {

//        noCallback();
        haveCallback();
//        syncApi();
    }

    //不带回调函数的 API
    private static void noCallback(){

        Properties properties = setConfigInfo();
        //创建生产者
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

        //发送数据
        for (int i = 1; i <= 4; i++) {

            ProducerRecord<String, String> record = new ProducerRecord<>("first",
                    Integer.toString(i), "xxxxxxxxxx");
            producer.send(record);
//            Thread.sleep(1000);
        }

//        producer.close();
        while (true){

        }
    }

    /**
     * 回调函数会在 producer 收到 ack 时调用，为异步调用，该方法有两个参数，分别是
     * RecordMetadata 和 Exception，如果 Exception 为 null，说明消息发送成功，如果
     * Exception 不为 null，说明消息发送失败。
     * 注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。
     */
    private static void haveCallback(){

        Properties properties = setConfigInfo();
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        //发送数据
        for (int i= 1; i<=4; i++){
            ProducerRecord<String, String> record = new ProducerRecord<>("first",
                    Integer.toString(i), "xxxxxxxxxx");
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e==null){
                        System.out.println("分区："+recordMetadata.partition()+"=====" + "游标："+recordMetadata.offset());
                    }else {
                        e.printStackTrace();
                    }

                }
            });
        }


        while (true){

        }

    }

    /**
     * 同步发送的意思就是，一条消息发送之后，会阻塞当前线程，直至返回 ack。
     * 由于 send 方法返回的是一个 Future 对象，根据 Futrue 对象的特点，我们也可以实现同
     * 步发送的效果，只需在调用 Future 对象的 get 方发即可。
     */
    private static void syncApi() throws ExecutionException, InterruptedException {

        Properties properties = setConfigInfo();
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        //发送数据
        for (int i= 1; i<= 4; i++){
            ProducerRecord<String, String> record = new ProducerRecord<>("first",
                    Integer.toString(i), "xxxxxxxxxx");
            producer.send(record).get();
        }


    }

    private static Properties setConfigInfo(){
        Properties properties = new Properties();
        //kafka集群
        properties.put("bootstrap.servers","212.129.155.35:9092");
        //ack
        properties.put("ack","all");
        //重试次数
        properties.put("retries",1);
        //批次大小  本版本是133  4条10字节的消息
        properties.put("batch.size",133);//只有数据积累到 batch.size 之后，sender线程才会发送数据
        //等待时间
        properties.put("linger.ms",10000);//如果数据迟迟未达到 batch.size，sender 等待 linger.time 之后就会发送数据 ms
        //RecordAccumulator 缓冲区大小
        properties.put("buffer.memory", 33554432);

        properties.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        return properties;

    }
}
